{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "b00d17d2-e967-43c5-8122-854a6c1af991",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "数据质量检验完成，报告已保存为'数据质量报告_.txt'\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import pandas as pd\n",
    "from typing import Dict, List, Tuple\n",
    "\n",
    "def extract_stock_code(filename: str) -> str:\n",
    "    \"\"\"从文件名中提取股票数字代码\"\"\"\n",
    "    if filename.startswith('sse.') or filename.startswith('szse.'):\n",
    "        return filename.split('.')[1]\n",
    "    return filename.split('.')[0]\n",
    "\n",
    "def map_at_to_ricequant(code: str) -> str:\n",
    "    \"\"\"将AT的股票代码格式映射为米筐格式\"\"\"\n",
    "    if code.startswith('6'):\n",
    "        return f\"{code}.XSHG\"\n",
    "    else:\n",
    "        return f\"{code}.XSHE\"\n",
    "\n",
    "def map_ricequant_to_at(code: str) -> str:\n",
    "    \"\"\"将米筐的股票代码格式映射为AT格式\"\"\"\n",
    "    prefix = code.split('.')[0]\n",
    "    exchange = code.split('.')[1]\n",
    "    if exchange == 'XSHG':\n",
    "        return f\"sse.{prefix}\"\n",
    "    else:\n",
    "        return f\"szse.{prefix}\"\n",
    "\n",
    "def compare_stock_lists(at_files: List[str], rq_files: List[str]) -> Tuple[List[str], List[str], List[str]]:\n",
    "    \"\"\"比较两个文件夹中的股票列表\"\"\"\n",
    "    at_codes = {extract_stock_code(f) for f in at_files}\n",
    "    rq_codes = {extract_stock_code(f) for f in rq_files}\n",
    "    \n",
    "    common_codes = at_codes & rq_codes\n",
    "    only_at = at_codes - rq_codes\n",
    "    only_rq = rq_codes - at_codes\n",
    "    \n",
    "    return sorted(common_codes), sorted(only_at), sorted(only_rq)\n",
    "\n",
    "def load_and_compare_data(at_path: str, rq_path: str, stock_code: str) -> pd.DataFrame:\n",
    "    \"\"\"加载并比较同一只股票在两个数据源中的数据\"\"\"\n",
    "    # 加载AT数据\n",
    "    at_file = f\"sse.{stock_code}.csv\" if stock_code.startswith('6') else f\"szse.{stock_code}.csv\"\n",
    "    at_df = pd.read_csv(os.path.join(at_path, at_file))\n",
    "    at_df['date'] = pd.to_datetime(at_df['time']).dt.strftime('%Y/%m/%d')\n",
    "    at_df = at_df[['date', 'open', 'high', 'low', 'close']]\n",
    "    at_df = at_df.rename(columns={\n",
    "        'open': 'AT_open',\n",
    "        'high': 'AT_high',\n",
    "        'low': 'AT_low',\n",
    "        'close': 'AT_close'\n",
    "    })\n",
    "    \n",
    "    # 加载米筐数据\n",
    "    rq_file = f\"{stock_code}.XSHG.csv\" if stock_code.startswith('6') else f\"{stock_code}.XSHE.csv\"\n",
    "    rq_df = pd.read_csv(os.path.join(rq_path, rq_file))\n",
    "    rq_df['date'] = pd.to_datetime(rq_df['date']).dt.strftime('%Y/%m/%d')\n",
    "    rq_df = rq_df[['date', 'open', 'high', 'low', 'close']]\n",
    "    rq_df = rq_df.rename(columns={\n",
    "        'open': 'RQ_open',\n",
    "        'high': 'RQ_high',\n",
    "        'low': 'RQ_low',\n",
    "        'close': 'RQ_close'\n",
    "    })\n",
    "    \n",
    "    # 合并数据\n",
    "    merged_df = pd.merge(at_df, rq_df, on='date', how='outer')\n",
    "    \n",
    "    # 添加缺失值标记\n",
    "    for field in ['open', 'high', 'low', 'close']:\n",
    "        merged_df[f'AT_{field}_missing'] = merged_df[f'AT_{field}'].isna()\n",
    "        merged_df[f'RQ_{field}_missing'] = merged_df[f'RQ_{field}'].isna()\n",
    "        merged_df[f'{field}_diff'] = abs(merged_df[f'AT_{field}'] - merged_df[f'RQ_{field}'])\n",
    "        merged_df[f'{field}_match'] = merged_df[f'{field}_diff'] < 1e-4\n",
    "    \n",
    "    return merged_df\n",
    "\n",
    "def generate_report(common_codes: List[str], only_at: List[str], only_rq: List[str], \n",
    "                   discrepancies: Dict[str, List[Dict[str, str]]]) -> str:\n",
    "    \"\"\"生成数据质量报告\"\"\"\n",
    "    report = []\n",
    "    \n",
    "    # 1. 股票列表比较结果\n",
    "    report.append(\"=\"*50)\n",
    "    report.append(\"股票列表比较结果\")\n",
    "    report.append(\"=\"*50)\n",
    "    report.append(f\"共同存在的股票数量: {len(common_codes)}\")\n",
    "    report.append(f\"仅存在于AT数据的股票: {len(only_at)}\")\n",
    "    if only_at:\n",
    "        report.append(\"股票代码: \" + \"、\".join(only_at))\n",
    "    report.append(f\"仅存在于米筐数据的股票: {len(only_rq)}\")\n",
    "    if only_rq:\n",
    "        report.append(\"股票代码: \" + \"、\".join(only_rq))\n",
    "    report.append(\"\\n\")\n",
    "\n",
    "    # 差异统计\n",
    "    report.append(\"\\n差异统计:\")\n",
    "    diff_stats = {\n",
    "        '总比对股票数': len(common_codes),\n",
    "        '完全匹配股票数': len(common_codes) - len(discrepancies),\n",
    "        '存在差异股票数': len(discrepancies),\n",
    "        '总差异记录数': sum(len(v) for v in discrepancies.values())\n",
    "    }\n",
    "    for k, v in diff_stats.items():\n",
    "        report.append(f\"{k}: {v}\")\n",
    "\n",
    "    # 2. 数据不一致详情\n",
    "    report.append(\"=\"*50)\n",
    "    report.append(\"数据不一致详情\")\n",
    "    report.append(\"=\"*50)\n",
    "    \n",
    "    if not discrepancies:\n",
    "        report.append(\"未发现数据不一致\")\n",
    "    else:\n",
    "        # 准备表格数据\n",
    "        table_data = []\n",
    "        for stock, issues in discrepancies.items():\n",
    "            for issue in issues:\n",
    "                table_data.append([\n",
    "                    stock,\n",
    "                    issue['date'],\n",
    "                    issue['field'],\n",
    "                    issue['at_value'],\n",
    "                    issue['rq_value']\n",
    "                ])\n",
    "        \n",
    "        # 创建表格\n",
    "        df = pd.DataFrame(table_data, columns=['股票代码', '日期', '字段', 'AT值', '米筐值'])\n",
    "        report.append(df.to_string(index=False, float_format=lambda x: f\"{x:.4f}\"))\n",
    "    \n",
    "    return \"\\n\".join(report)\n",
    "\n",
    "def main():\n",
    "    # 文件夹路径\n",
    "    at_path = r\"C:\\Users\\86152\\Desktop\\A股日线数据\\AT的A股日线\"\n",
    "    rq_path = r\"C:\\Users\\86152\\Desktop\\A股日线数据\\米筐的A股日线\"\n",
    "    \n",
    "    # 获取文件列表\n",
    "    at_files = [f for f in os.listdir(at_path) if f.endswith('.csv')]\n",
    "    rq_files = [f for f in os.listdir(rq_path) if f.endswith('.csv')]\n",
    "    \n",
    "    # 比较股票列表\n",
    "    common_codes, only_at, only_rq = compare_stock_lists(at_files, rq_files)\n",
    "    \n",
    "    # 检查数据不一致\n",
    "    discrepancies = {}\n",
    "    for code in common_codes[:]:  # 为了演示，只检查前10只股票\n",
    "        try:\n",
    "            merged_df = load_and_compare_data(at_path, rq_path, code)\n",
    "            issues = []\n",
    "        \n",
    "            for _, row in merged_df.iterrows():\n",
    "                for field in ['open', 'high', 'low', 'close']:\n",
    "                    at_missing = row[f'AT_{field}_missing']\n",
    "                    rq_missing = row[f'RQ_{field}_missing']\n",
    "                \n",
    "                    if at_missing or rq_missing:\n",
    "                        issues.append({\n",
    "                            'date': row['date'],\n",
    "                            'field': f\"{field}(缺失)\",\n",
    "                            'at_value': \"缺失\" if at_missing else \"存在\",\n",
    "                            'rq_value': \"缺失\" if rq_missing else \"存在\"\n",
    "                        })\n",
    "                    elif not row[f'{field}_match']:\n",
    "                        diff_pct = abs(row[f'AT_{field}'] - row[f'RQ_{field}'])/row[f'RQ_{field}']*100\n",
    "                        issues.append({\n",
    "                            'date': row['date'],\n",
    "                            'field': field,\n",
    "                            'at_value': row[f'AT_{field}'],\n",
    "                            'rq_value': row[f'RQ_{field}'],\n",
    "                            '差异百分比': f\"{diff_pct:.2f}%\"\n",
    "                        })\n",
    "        \n",
    "            if issues:\n",
    "                discrepancies[code] = issues\n",
    "            \n",
    "        except Exception as e:\n",
    "            print(f\"处理股票 {code} 时出错: {str(e)}\")\n",
    "    \n",
    "    # 生成并保存报告\n",
    "    report = generate_report(common_codes, only_at, only_rq, discrepancies)\n",
    "    with open('数据质量报告_.txt', 'w', encoding='utf-8') as f:\n",
    "        f.write(report)\n",
    "    \n",
    "    print(\"数据质量检验完成，报告已保存为'数据质量报告_.txt'\")\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a17cbc23-f463-4012-95b2-568f60dbad9d",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "deaecb02-61e4-436a-bb62-b12b6cf0e550",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bee997de-037c-42a0-ac6d-a718a42d2c5d",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "74c4489c-0169-4ffe-9a5d-9cd8dc5d435f",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0ef50c37-4fe6-4646-8402-af3f3512af9c",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
